package com.bfxy.rabbit.producer.autoconfigure.broker;

import com.bfxy.base.rabbit.api.Message;
import com.bfxy.base.rabbit.api.MessageType;
import com.bfxy.base.rabbit.common.convert.GenericMessageConverter;
import com.bfxy.base.rabbit.common.convert.RabbitMessageConverter;
import com.bfxy.base.rabbit.common.serializer.Serializer;
import com.bfxy.base.rabbit.common.serializer.SerializerFactory;
import com.bfxy.base.rabbit.common.serializer.impl.JacksonSerializerFactory;
import com.google.common.base.Preconditions;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * RabbitTemplateContainer池化封装
 * @author linxin
 */
@Slf4j
@Component
public class RabbitTemplateContainer implements RabbitTemplate.ConfirmCallback {

    private Map<String /* TOPIC */, RabbitTemplate> rabbitMap = Maps.newConcurrentMap();
    @Autowired
    private ConnectionFactory connectionFactory;
    private SerializerFactory serializerFactory = JacksonSerializerFactory.INSTANCE;

    public RabbitTemplate getTemplate(Message message){
        Preconditions.checkNotNull(message);
        String topic = message.getTopic();
        RabbitTemplate rabbitTemplate = rabbitMap.get(topic);
        if(rabbitTemplate != null) {
            return rabbitTemplate;
        }
        log.info("#RabbitTemplateContainer.getTemplate# topic: {} is not exists, create one", topic);
        RabbitTemplate newTemplate = new RabbitTemplate(connectionFactory);
        newTemplate.setExchange(topic);
        newTemplate.setRoutingKey(message.getRoutingKey());
        newTemplate.setRetryTemplate(new RetryTemplate());

        //	添加序列化反序列化和converter对象
        Serializer serializer = serializerFactory.create();
        GenericMessageConverter gmc = new GenericMessageConverter(serializer);
        RabbitMessageConverter rmc = new RabbitMessageConverter(gmc);
        newTemplate.setMessageConverter(rmc);

        String messageType = message.getMessageType();
        if(!MessageType.RAPID.equals(messageType)) {
            newTemplate.setConfirmCallback(this);
        }
        rabbitMap.putIfAbsent(topic, newTemplate);
        return rabbitMap.get(topic);
    }
    @Override
    public void confirm(CorrelationData correlationData, boolean b, String s) {

    }
}
